package com.acme.concurrent.executor;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author Gregory Sobko
 */
class FileQueuePool {
    // POISONOUS PILL, IF INDEXER GETS ONE - IT SHOULD DIE!
    public static final File POISONOUS_PILL = new File("/dev/null");
    private CountDownLatch doneSignal;
    private List<BlockingDeque<File>> pool;
    private final int count;
    private AtomicInteger current = new AtomicInteger(0);
    private static final int BOUND = 10;
    private volatile boolean isShutDown = false;

    FileQueuePool(final int consumerCount, final int producersCount) {
        this.count = consumerCount;
        this.pool = new ArrayList<BlockingDeque<File>>(count);
        for (int i = 0; i < count; i++) {
            pool.add(i, new LinkedBlockingDeque<File>(BOUND));
        }
        doneSignal = new CountDownLatch(producersCount);
    }

    public BlockingQueue<File> getQueue(int index){
        return pool.get(index);
    }

    public File trySteal(int index){
        for (int i = index + 1; i < index + count; i++) {
            // Check nothing appeared in our own Q
            if(pool.get(index).size() > 0){
                return null;
            }
            // Ok, let's poll element from our neighbour
            File file = pool.get(i % count).pollFirst();
            // Check if we've stolen poisonous pill
            if(file == POISONOUS_PILL){
                // Oh, that's baaad! We've stolen poisonous pill =(
                // Let's put it back fast!
                // while (true) - is to make assure some other bud was poisoned
                while (true){
                    try {
                        pool.get(i % count).put(POISONOUS_PILL);
                        // Ok, we put pill back, let's exit, because there should be pill in our own Q
                        return null;
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
            if(file != null){
                return file;
            }
        }
        return null;
    }

    public void put(File file) throws InterruptedException {
        if(!isShutDown){
            pool.get(current.getAndIncrement() % count).put(file);
        }else{
            throw new IllegalStateException("put(...) method is not acceptable! Queue is closed!");
        }
    }

    public void signalProducerDone(){
        doneSignal.countDown();
    }

    public void awaitFinalization() throws InterruptedException {
        doneSignal.await();
    }

    public void awaitFinalization(long timeout, TimeUnit timeUnit) throws InterruptedException {
        doneSignal.await(timeout, timeUnit);
    }

    public void shutdown() {
        this.isShutDown = true;
        int i = 0;
        while (i < count) {
            try {
                pool.get(i).put(POISONOUS_PILL);
                i++;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}